Join Strategies

While specifying the type of join (Inner, Outer, etc.) determines the logical outcome, Spark must choose a physical join strategy to execute the join across the cluster. Understanding these physical strategies is critical to debugging performance bottlenecks and out-of-memory errors in Spark.

The Core Join Strategies

Spark uses three primary physical join strategies under the hood:

graph TD
    subgraph JoinStrategies["Spark Join Strategies"]
        direction TB
        SMJ["Sort-Merge Join (SMJ)<br>- Shuffles and sorts both datasets<br>- Standard for huge-to-huge tables"]
        BHJ["Broadcast Hash Join (BHJ)<br>- Broadcasts small table to all executors<br>- Fast, completely skips shuffles"]
        SHJ["Shuffle Hash Join<br>- Shuffles but doesn't sort<br>- Uses hash tables locally"]
    end
    style JoinStrategies fill:#eef2f6,stroke:#475569,stroke-width:2px;

1. Sort-Merge Join (SMJ)

• The Default Strategy: Used when both joining datasets are large.
• How it works:
  1. Shuffle Phase: Both tables are hashed and shuffled across the network based on the join key, ensuring rows with matching keys land in the same partition on the same executor.
  2. Sort Phase: Rows inside each partition are sorted by the join key.
  3. Merge Phase: The sorted partitions are joined by scanning through the records line-by-line (which is highly efficient since they are pre-sorted).
• Performance Cost: Very High. Network shuffling and sorting are highly disk and CPU-intensive.

2. Broadcast Hash Join (BHJ)

• The Speed King: Used when one of the joining tables is small (default threshold is 10 MB or less).
• How it works:
  1. The small DataFrame is downloaded fully to the Driver node.
  2. The Driver broadcasts a copy of this small table as a hash map to all worker executors on the cluster.
  3. Each executor performs a fast local lookup join as it scans its own partition of the large table.
• Performance Cost: Zero Shuffle! Network shuffles are completely avoided, converting wide dependencies into high-speed narrow dependencies.

3. Shuffle Hash Join

• Similar to Sort-Merge, but it builds local hash tables on the partitions instead of sorting. Used when data is skewed or sorting cannot be done efficiently.

PySpark Code Example: Forcing a Broadcast Join

If you know a table is small (e.g. less than 100 MB), you can manually instruct Spark to broadcast it using pyspark.sql.functions.broadcast():

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast, col

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("Join Strategies") \
    .master("local[*]") \
    .getOrCreate()

# 2. Large Transactions DataFrame (Fact Table)
tx_data = [(101, 1, 500.0), (102, 2, 20.0), (103, 1, 150.0)]
tx_df = spark.createDataFrame(tx_data, ["tx_id", "user_id", "amount"])

# 3. Small Users DataFrame (Dimension Table - Less than 50MB)
users_data = [(1, "Alice"), (2, "Bob")]
users_df = spark.createDataFrame(users_data, ["user_id", "user_name"])

# 4. Perform Join forcing a Broadcast Join on users df
# This avoids shuffles on both datasets entirely!
broadcast_joined_df = tx_df.join(broadcast(users_df), "user_id", "inner")
broadcast_joined_df.show()

# 5. Inspect the Physical Execution Plan
# Look for 'BroadcastHashJoin' vs 'SortMergeJoin' in the output!
broadcast_joined_df.explain()